package com.group1.realtime.app.dws;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.group1.realtime.app.BaseAppV2;
import com.group1.realtime.bean.TrafficPageViewBean;
import com.group1.realtime.common.Constant;
import com.group1.realtime.util.AtguiguUtil;
import com.group1.realtime.util.FlinkSinkUtil;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.time.Duration;
import java.util.HashMap;

public class Dws_TrafficPageViewWindow extends BaseAppV2 {

    public static void main(String[] args) {
        new Dws_TrafficPageViewWindow().init(
                3002,
                2,
                "Dws_TrafficPageViewWindow",
                Constant.TOPIC_DWD_TRAFFIC_PAGE,
                Constant.TOPIC_DWD_TRAFFIC_UV_DETAIL,
                Constant.TOPIC_DWD_TRAFFIC_UJ_DETAIL
        );
    }

    @Override
    protected void handle(StreamExecutionEnvironment env, HashMap<String, DataStreamSource<String>> topicAndStreamMap) {

        DataStream<TrafficPageViewBean> beanStream = parseAndUnionOne(topicAndStreamMap);
        SingleOutputStreamOperator<TrafficPageViewBean> reducedStream = windowAndAgg(beanStream);
        writeClickHouse(reducedStream);
    }

    private void writeClickHouse(SingleOutputStreamOperator<TrafficPageViewBean> reducedStream) {
        reducedStream.addSink(FlinkSinkUtil.getClickHouseSink("dws_traffic_page_view_window",TrafficPageViewBean.class));
    }

    private SingleOutputStreamOperator<TrafficPageViewBean> windowAndAgg(DataStream<TrafficPageViewBean> beanStream) {

        return beanStream
                .keyBy(bean->bean.getVc()+"_"+bean.getCh()+"_"+bean.getAr()+"_"+bean.getIsNew())
                .window(TumblingEventTimeWindows.of(Time.seconds(5)))
                .reduce(new ReduceFunction<TrafficPageViewBean>() {
                            @Override
                            public TrafficPageViewBean reduce(TrafficPageViewBean t1,
                                                              TrafficPageViewBean t2) throws Exception {
                                t1.setUjCt(t1.getUjCt() + t2.getUjCt());
                                t1.setUvCt(t1.getUvCt() + t2.getUvCt());
                                t1.setSvCt(t1.getSvCt() + t2.getSvCt());
                                t1.setDurSum(t1.getDurSum() + t2.getDurSum());
                                t1.setPvCt(t1.getPvCt() + t2.getPvCt());
                                return t1;
                            }
                        },
                        new ProcessWindowFunction<TrafficPageViewBean, TrafficPageViewBean, String, TimeWindow>() {
                            @Override
                            public void process(String key,
                                                Context context,
                                                Iterable<TrafficPageViewBean> iterable,
                                                Collector<TrafficPageViewBean> out) throws Exception {
                                TrafficPageViewBean next = iterable.iterator().next();
                                next.setStt(AtguiguUtil.toDateTime(context.window().getStart()));
                                next.setEdt(AtguiguUtil.toDateTime(context.window().getEnd()));

                                next.setTs(context.currentProcessingTime());


                                out.collect(next);
                            }
                        });

    }

    private DataStream<TrafficPageViewBean> parseAndUnionOne(HashMap<String, DataStreamSource<String>> topicAndStreamMap) {
        SingleOutputStreamOperator<TrafficPageViewBean> pageStram = topicAndStreamMap
                .get(Constant.TOPIC_DWD_TRAFFIC_PAGE)
                .map(new MapFunction<String, TrafficPageViewBean>() {
                    @Override
                    public TrafficPageViewBean map(String data) throws Exception {
                        JSONObject jsonObject = JSON.parseObject(data);
                        JSONObject common = jsonObject.getJSONObject("common");
                        JSONObject page = jsonObject.getJSONObject("page");
                        Long ts = jsonObject.getLong("ts");


                        return TrafficPageViewBean.builder()
                                .vc(common.getString("vc"))
                                .ch(common.getString("ch"))
                                .ar(common.getString("ar"))
                                .sc(common.getString("sc"))
                                .isNew(common.getString("is_new"))
                                .uvCt(0L)
                                .svCt(page.getString("last_page_id") == null ? 1L : 0L)
                                .pvCt(1L)
                                .durSum(page.getLong("during_time") == null ? 0L : page.getLong("during_time"))
                                .ujCt(0L)
                                .ts(ts)
                                .build();
                    }
                }).assignTimestampsAndWatermarks(
                        WatermarkStrategy
                                .<TrafficPageViewBean>forBoundedOutOfOrderness(Duration.ofSeconds(5))
                                .withTimestampAssigner((bean,ts)->bean.getTs())
                                .withIdleness(Duration.ofSeconds(10))
                );


        SingleOutputStreamOperator<TrafficPageViewBean> uvStream = topicAndStreamMap
                .get(Constant.TOPIC_DWD_TRAFFIC_UV_DETAIL)
                .map(new MapFunction<String, TrafficPageViewBean>() {
                    @Override
                    public TrafficPageViewBean map(String data) throws Exception {
                        JSONObject jsonObject = JSON.parseObject(data);
                        JSONObject common = jsonObject.getJSONObject("common");
                        JSONObject page = jsonObject.getJSONObject("page");
                        Long ts = jsonObject.getLong("ts");

                        return TrafficPageViewBean.builder()
                                .vc(common.getString("vc"))
                                .ch(common.getString("ch"))
                                .ar(common.getString("ar"))
                                .sc(common.getString("sc"))
                                .isNew(common.getString("is_new"))
                                .uvCt(1L)
                                .svCt(0L)
                                .pvCt(1L)
                                .durSum(0L)
                                .ujCt(0L)
                                .ts(ts)
                                .build();
                    }
                })
                .assignTimestampsAndWatermarks(
                        WatermarkStrategy
                                .<TrafficPageViewBean>forBoundedOutOfOrderness(Duration.ofSeconds(5))
                                .withTimestampAssigner((bean,ts)->bean.getTs())
                                .withIdleness(Duration.ofSeconds(10))
                );


        SingleOutputStreamOperator<TrafficPageViewBean> ujStream = topicAndStreamMap
                .get(Constant.TOPIC_DWD_TRAFFIC_UJ_DETAIL)
                .map(new MapFunction<String, TrafficPageViewBean>() {
                    @Override
                    public TrafficPageViewBean map(String data) throws Exception {
                        JSONObject jsonObject = JSON.parseObject(data);
                        JSONObject common = jsonObject.getJSONObject("common");
                        JSONObject page = jsonObject.getJSONObject("page");
                        Long ts = jsonObject.getLong("ts");

                        return TrafficPageViewBean.builder()
                                .vc(common.getString("vc"))
                                .ch(common.getString("ch"))
                                .ar(common.getString("ar"))
                                .sc(common.getString("sc"))
                                .isNew(common.getString("is_new"))
                                .uvCt(0L)
                                .svCt(0L)
                                .pvCt(0L)
                                .durSum(0L)
                                .ujCt(1L)
                                .ts(ts)
                                .build();
                    }
                })
                .assignTimestampsAndWatermarks(
                        WatermarkStrategy
                                .<TrafficPageViewBean>forBoundedOutOfOrderness(Duration.ofSeconds(5))
                                .withTimestampAssigner((bean,ts)->bean.getTs())
                                .withIdleness(Duration.ofSeconds(10))
                );

        return pageStram.union(uvStream,ujStream);
    }
}
